package priv.mill.hdfs.impl;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import priv.mill.hdfs.api.HdfsApi;
import priv.mill.hdfs.api.HdfsFileHandler;

@Component
public class HdfsUtil implements HdfsApi{

	
	@Autowired
	private FileSystem fileSystem;
	
	
	@Override
	public List<String> listNames(String path) throws Exception {
		List<FileStatus> listStatus = list(path);
		List<String> names = new ArrayList<>();
		for (FileStatus fileStatus : listStatus) {
			names.add(fileStatus.getPath().getName());
		}
		return names;
	}


	
	

	public FileSystem getFileSystem() {
		return fileSystem;
	}

	public void setFileSystem(FileSystem fileSystem) {
		this.fileSystem = fileSystem;
	}


	@Override
	public List<FileStatus> list(String path) throws Exception {
		FileStatus[] listStatus = fileSystem.listStatus(new Path(path));
		return new ArrayList<>(Arrays.asList(listStatus));
	}





	@Override
	public void list(String path, HdfsFileHandler hdfsFileHandler) throws Exception {
		RemoteIterator<LocatedFileStatus> listFiles = fileSystem.listFiles(new Path(path), true);
		while(listFiles.hasNext()) {
			LocatedFileStatus next = listFiles.next();
			hdfsFileHandler.doHandler(next.getPath(),next.getBlockLocations(),next);
		}
	}





	@Override
	public boolean mkdir(String path) throws Exception {
		if(fileSystem.exists(new Path(path)))
			return false;
		return fileSystem.mkdirs(new Path(path));
	}





	@Override
	public byte[] get(String string) throws Exception {
		try(FSDataInputStream open = fileSystem.open(new Path(string));){
			ByteArrayOutputStream bOut = new ByteArrayOutputStream();
			byte[] b = new byte[4096];
			int len;
			while((len=open.read(b))>0) {
				bOut.write(b, 0, len);
			}
			return bOut.toByteArray();
		}
	}





	@Override
	public boolean exists(String string) throws Exception {
		return fileSystem.exists(new Path(string));
	}





	@Override
	public void createFile(String path, byte[] bytes) throws Exception {
		try(
				FSDataOutputStream create = fileSystem.create(new Path(path));
				InputStream in = new ByteArrayInputStream(bytes);){
			
			IOUtils.copyBytes(in, create, bytes.length, true);
		}
		
	}





	@Override
	public void put(String src, String dist) throws Exception {
		fileSystem.copyFromLocalFile(new Path(src), new Path(dist));
	}





	@Override
	public boolean delete(String path) throws Exception {
		return fileSystem.delete(new Path(path),true);
	}





	@Override
	public boolean rename(String src, String dist) throws Exception {
		return fileSystem.rename(new Path(src), new Path(dist));
	}





	@Override
	public void getFile(String hdfsPath, String localPath) throws Exception {
		try (FSDataInputStream open = fileSystem.open(new Path(hdfsPath));){
			File file = new File(localPath);
			if (file.isFile() && file.exists()) {
				throw new IOException("本地文件已存在");
			}
			if(file.isDirectory()) {
				java.nio.file.Path path = Paths.get(localPath);
				File file2 = new File(hdfsPath);
				String name = file2.getName();
				file = path.resolve(name).toFile();
			}
			
			try(FileOutputStream fileOutputStream = new FileOutputStream(file);){
				IOUtils.copyBytes(open, fileOutputStream, open.available(), true);
			}
		} catch (Exception e) {
			throw e;
		}
	}





}
